package juc.blockqueue.arrayblockqueue;

/**
 * @ClassName DataPipeline
 * @PackageName juc.blockqueue.arrayblockqueue
 * @projectName myThread
 * @Description 4. 数据处理流水
 * @Author games
 * @Date 2025/9/4 上午10:43
 * @Version 1.0
 */
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * 4. 数据处理流水线
 */
public class DataPipeline {
    public static void main(String[] args) {
        BlockingQueue<Integer> rawQueue = new ArrayBlockingQueue<>(5);
        BlockingQueue<Integer> processedQueue = new ArrayBlockingQueue<>(5);

        // 数据生成阶段
        Thread dataGenerator = new Thread(() -> {
            try {
                for (int i = 1; i <= 10; i++) {
                    rawQueue.put(i);
                    System.out.println("Generated data: " + i);
                    Thread.sleep(300);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        // 数据处理阶段
        Thread dataProcessor = new Thread(() -> {
            try {
                while (true) {
                    Integer data = rawQueue.take();
                    Integer processedData = data * data; // 简单处理：平方
                    processedQueue.put(processedData);
                    System.out.println("Processed data: " + data + " -> " + processedData);
                    Thread.sleep(500);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        // 数据输出阶段
        Thread dataOutput = new Thread(() -> {
            try {
                while (true) {
                    Integer data = processedQueue.take();
                    System.out.println("Output result: " + data);
                    Thread.sleep(700);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        dataGenerator.start();
        dataProcessor.start();
        dataOutput.start();
    }
}
